package com.yunc.common.disruptor;

import cn.hutool.core.collection.CollUtil;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.yunc.common.core.utils.ProxyUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;

@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DisruptorService implements InitializingBean {

    private final ApplicationContext applicationContext;

    private Disruptor<Event> disruptor;
    private static final int RING_BUFFER_SIZE = 1024 * 1024;

    public <T> void publishEvent(T event) {
        Event commEvent = new Event(event);
        final long seq = disruptor.getRingBuffer().next();
        Event userEvent = disruptor.get(seq);
        userEvent.setValue(commEvent.getValue());
        disruptor.getRingBuffer().publish(seq);
    }

    @Override
    public void afterPropertiesSet() {
        // 获取所有bean
        if (applicationContext == null) {
            return;
        }
        Map<String, Object> map = applicationContext.getBeansWithAnnotation(Handler.class);
        Set<EventHandler> eventHandlers = new HashSet<>();
        for (Map.Entry<String, Object> entrymap : map.entrySet()) {
            try {
                // 通过反射获取相关的实现类的Object
                Object object = ProxyUtil.getTarget(entrymap.getValue());
                if (object != null) {
                    eventHandlers.add((EventHandler) object);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        disruptor = new Disruptor<>(Event.FACTORY, RING_BUFFER_SIZE, Executors.defaultThreadFactory(), ProducerType.SINGLE,new BlockingWaitStrategy());
        // 手动添加handler
        if (CollUtil.isNotEmpty(eventHandlers)) {
            for (EventHandler handler : eventHandlers) {
                // if (filterHandler(handler)) {
                disruptor.handleEventsWith(handler);
            }
        }
        disruptor.start();
    }
}
